//
//  IOMonad.swift
//  iOSDemo
//
//  Created by FH on 2020/9/30.
//  Copyright © 2020 Fuhan. All rights reserved.
//

import Foundation

/**
 stream中的数据，可以把stream看做通用的范畴(Box)内数据的抽象形式
 
 @param value 表示stream流动的数据，如果为nil表示忽略本次数据
 @param isComplete 表示当前的stream是否终止
 */
public typealias StreamValue<T> = (value: T?, isComplete: Bool)
/** 函子通用的转换函数声明 */
public typealias TransformFn<T, U> = (T) -> StreamValue<U>
/** 函子通用的Bind函数声明 */
public typealias FunctorBindFn<T, U> = () -> TransformFn<T, U>
/**
 单子通用Bind函数声明
 TODO: 去掉curry，目前看没有用处。curry化的目的是在wrapper函子时，一些函子的实现需要closure参数
 */
public typealias MonadBindFn<T, U> = () -> TransformFn<T, IOMonad<U>>
/** 单子封装异步IO Thunk的函数声明 */
public typealias MonadIOThunk<T> = (IOThunker<T>) -> Action?


/////////////////////////////////////////

/*
 TODO:
 1. debugAllPaths方法可以打印出从第一个Monad实例开始所有的函子路径
 2. schedule的deliveOn/subscribeOn与functor list相结合
    a. queue通过set_target来提升优先级，以及设置关联。比如关联到mainQueue、将多个queue关联到serialQueue后，这些queue的async就变成顺序执行
    b. subscribeOn没什么用，deliverOn就够用了，就是简单的包装后做async()调用，需要注意isDispose与Schedule的联动
 */


/** 用于清理IO资源的Disposer */
public class IODisposer : Copyable {
    private var disposeFn: Action? = nil
    fileprivate var isDisposed: Bool = false
    
    required init(instance: IODisposer) {
        self.disposeFn = instance.disposeFn
    }
    
    init() { }
    
    func setDispose(_ disposeFn: @escaping Action) {
        // 对于同步的Thunker，在subscribe完成后对应的disposer已经结束
        if self.isDisposed {
            disposeFn()
        } else {
            self.disposeFn = disposeFn
        }
    }
    
    func dispose() {
        // double check
        if !self.isDisposed {
            if let disposeFn = self.disposeFn {
                disposeFn()
            }
            self.isDisposed = true
        }
    }
}

//////////////////////////////////////////

public class IOThunker<T> : Copyable {
    /** 当前Thunker已经被订阅，用于copy-on-write */
    fileprivate var isSubscribed = false
    /** 当前所归属的Monad名字，用于调试 */
    fileprivate var debugName: String = ""
    /** 绑定的IO ThunkFn */
    private let thunk: MonadIOThunk<T>
    /** 绑定的IO Disposer，利用disposer.isDisposed判定Stream是否终止 */
    private let disposer: IODisposer
    /** 外部订阅的NextFn */
    private var onNextFn: Action1<T>! = nil
    /** 外部订阅的CompleteFn */
    private var onCompleteFn: Action! = nil
    /** 外部订阅的ErrorFn */
    private var onErrorFn: Action1<Error>! = nil
    /** 绑定的函子列表 */
    private var bindFunctors: [TransformFn<T, T>] = []
    /** 外部赋予的初始值，非IO产生 */
    private var startValue: T? = nil
    
    required init(instance: IOThunker) {
        self.debugName = instance.debugName
        self.thunk = instance.thunk
        self.startValue = instance.startValue
        self.disposer = instance.disposer.copy()
    }
    
    fileprivate init(thunk: @escaping MonadIOThunk<T>) {
        self.thunk = thunk
        self.disposer = IODisposer()
    }
    
    fileprivate func setDebugName(name: String) {
        self.debugName = name
    }
    
    /** 绑定一种函子，减少不必要的closure回调，将不同的函子绑定到一起操作 */
    fileprivate func binding(functor: @escaping TransformFn<T, T>) {
        self.bindFunctors.append(functor)
    }
    
    /** 用于外部定于IO Stream的Result */
    fileprivate func subscribe(nextFn: @escaping Action1<T>,
                               completeFn: @escaping Action,
                               errorFn: @escaping Action1<Error>) -> IODisposer {
        self.isSubscribed = true
        self.onNextFn = nextFn
        self.onCompleteFn = completeFn
        self.onErrorFn = errorFn
        
        // 激活IO，冷 -> 热
        if let disposeFn = self.thunk(self) {
            if let startValue = self.startValue {
                _sendNext(startValue)
                self.startValue = nil
            }
            self.disposer.setDispose(disposeFn)
        }
        return self.disposer
    }
    
    private func _sendNext(_ val: T) {
        var isBreak = false
        var nextVal: T? = val
        if !self.disposer.isDisposed {
            var willComplete = false
            for functor in self.bindFunctors {
                let (valOrNil, isComplete) = functor(nextVal!)
                // 修复valOrNil == nil无效问题
                nextVal = valOrNil
                // 当前有值则发送
                if nextVal == nil {
                    // 当前函子操作后，忽略本次sendNext
                    isBreak = true
                }
                // 当前函子操作后，整个Stream结束
                if isComplete {
                    willComplete = true
                    isBreak = true
                }
                
                // double check
                if self.disposer.isDisposed {
                    isBreak = true
                    willComplete = false
                    nextVal = nil
                }
                if isBreak {
                    break
                }
            }
            if let val = nextVal {
                self.onNextFn(val)
            }
            if willComplete {
                self.sendComplete()
            }
        }
    }
    
    public func setStartValue(val: T) {
        self.startValue = val
    }
    
    public func sendNext(_ val: T) {
        if !self.disposer.isDisposed {
            _sendNext(val)
        }
    }
    
    public func sendComplete() {
        if !self.disposer.isDisposed {
            self.onCompleteFn()
            self.disposer.dispose()
        }
    }
    
    public func sendError(_ err: Error) {
        if !self.disposer.isDisposed {
            self.onErrorFn(err)
            self.disposer.dispose()
        }
    }
}

private class ThunkerShim<T> : Copyable {
    /** 默认为引用语义 */
    var isValueSemantics = false
    /** 当前的thunker，即thunker */
    var thunker: IOThunker<T>
    /** 未curry化的准备绑定的Functor列表，用于复制thunker */
    var bindFunctorList: [FunctorBindFn<T, T>] = []

    required init(instance: ThunkerShim) {
        self.bindFunctorList = instance.bindFunctorList
        self.thunker = Self.copy(thunker: instance.thunker, bindFunctorList: self.bindFunctorList)
        self.isValueSemantics = instance.isValueSemantics
    }
    
    init(thunk: @escaping MonadIOThunk<T>) {
        self.thunker = IOThunker(thunk: thunk)
    }
    
    private static func copy(thunker: IOThunker<T>, bindFunctorList: [FunctorBindFn<T, T>]) -> IOThunker<T> {
        let copyThunker = thunker.copy()
        
        for bindFn in bindFunctorList {
            let transformFn = bindFn()
            copyThunker.binding(functor: transformFn)
        }
        return copyThunker
    }
}

public protocol AnyIOStream {
    @discardableResult
    func subscribeAny(nextFn: @escaping Action1<Any>,
                      completeFn: @escaping Action,
                      errorFn: @escaping Action1<Error>) -> IODisposer
    
    func mapAny() -> IOMonad<Any>
}

public struct IOMonad<T> {
    private var shim: ThunkerShim<T>
    
    private init(shim: ThunkerShim<T>) {
        self.shim = shim
    }
    
    public init(_ thunk: @escaping MonadIOThunk<T>) {
        self.shim = ThunkerShim(thunk: thunk)
    }
    
    /**
     开启值语义
     1. 标准的Rac/Rx对bind的所有实现(functor/monad)都是closure闭包/嵌套/curry化的，而这种形式就是值语义的
     2. 当前实现为了减少closure嵌套，所以对functor的bind是list的，不具备值语义。也无法通过某些调用做辅助判定
     3. 但在实际应用中使用值语义的场景大多为重复subscribe且不常见，为此当前实现默认为引用语义，必要时启用值语义
     */
    @discardableResult
    public func enableValueSemantics() -> IOMonad {
        self.shim.isValueSemantics = true
        return self
    }
    
    public func debugName(_ name: String) -> IOMonad {
        self.shim.thunker.setDebugName(name: name)
        return self
    }
    
    // TODO: debugName arg
    public func bindFunctor(_ binNextdFn: @escaping FunctorBindFn<T, T>) -> IOMonad<T> {
        // copy-on-write，第一个被订阅不需要copy
        let shim = (self.shim.isValueSemantics
                    && self.shim.thunker.isSubscribed)
            ? self.shim.copy()
            : self.shim
        assert(!shim.thunker.isSubscribed, "please enable Value Semantics.")
        
        shim.bindFunctorList.append(binNextdFn)
        // 获得curry化后的函数实例
        let transformFn = binNextdFn()
        shim.thunker.binding(functor: transformFn)
        if (self.shim.isValueSemantics) {
            return IOMonad(shim: shim)
        } else {
            return self
        }
    }
    
    public func bindMonad<U>(_ bindNextFn: @escaping MonadBindFn<T, U>) -> IOMonad<U> {
        return IOMonad<U> { thunker in
            let mutex = Mutex()
            // 对应当前Stream的Id
            let selfId = UUID()
            // 记录所有的Id，用于判断是否所有Stream全部完成
            var recordIds = [selfId]
            // 组合所有Stream的Disposer
            let composeDisposer = IODisposer()
            // 记录所有Stream返回的Disposer
            var allDisposer: [IODisposer] = []
            
            // 等待所有子Stream完成的函数
            let allCompleteFn = { (streamId: UUID) in
                var remainCount = 0
                let succ = mutex.tryLock()
                recordIds.removeAll { $0 == streamId }
                remainCount = recordIds.count
                if succ {
                    mutex.unlock()
                }
                if remainCount == 0 {
                    thunker.sendComplete()
                }
            }
            
        
            let transformFn = bindNextFn()
            let preDisposer = self.enableValueSemantics().subscribe(nextFn: { val in
                let (streamOrNil, isComplete) = transformFn(val)
                if let stream = streamOrNil {
                    // 获得stream的Id
                    let streamId = UUID()
                    var succ = mutex.tryLock()
                    recordIds.append(streamId)
                    if succ {
                        mutex.unlock()
                    }
                    // 订阅stream
                    let streamDisposer = stream.enableValueSemantics().subscribe(nextFn: { streamVal in
                        thunker.sendNext(streamVal)
                    }, completeFn: {
                        allCompleteFn(streamId)
                    }) { streamErr in
                        thunker.sendError(streamErr)
                    }
                    
                    succ = mutex.tryLock()
                    if composeDisposer.isDisposed {
                        streamDisposer.dispose()
                    } else {
                        allDisposer.append(streamDisposer)
                    }
                    if succ {
                        mutex.unlock()
                    }
                }
                if isComplete {
                    allCompleteFn(selfId)
                }
            }, completeFn: {
                allCompleteFn(selfId)
            }) { err in
                thunker.sendError(err)
            }
            
            composeDisposer.setDispose {
                let succ = mutex.tryLock()
                for disposer in allDisposer {
                    disposer.dispose()
                }
                preDisposer.dispose()
                if succ {
                    mutex.unlock()
                }
            }
            
            return composeDisposer.dispose
        }
    }
    
    @discardableResult
    public func subscribe(nextFn: @escaping Action1<T> = { _ in },
                          completeFn: @escaping Action = { },
                          errorFn: @escaping Action1<Error> = { _ in }) -> IODisposer {
        // copy-on-write，第一个被订阅不需要copy
        let shim = (self.shim.isValueSemantics
                    && self.shim.thunker.isSubscribed)
            ? self.shim.copy()
            : self.shim
        assert(!shim.thunker.isSubscribed, "please enable Value Semantics.")
        
        return shim.thunker.subscribe(nextFn: nextFn, completeFn: completeFn, errorFn: errorFn)
    }
    
    /** 设置默认值 */
    public func start(with val: T) -> IOMonad {
        assert(!self.shim.thunker.isSubscribed, "please enable Value Semantics.")
        
        let shim = self.shim.isValueSemantics ? self.shim.copy() : self.shim
        shim.thunker.setStartValue(val: val)
        if (self.shim.isValueSemantics) {
            return IOMonad(shim: shim)
        } else {
            return self
        }
    }
}

extension IOMonad : AnyIOStream {
    @discardableResult
    public func subscribeAny(nextFn: @escaping Action1<Any> = { _ in },
                             completeFn: @escaping Action = { },
                             errorFn: @escaping Action1<Error> = { _ in }) -> IODisposer {
        return self.map { val -> Any in
            return val
        }.subscribe(nextFn: { val in
            nextFn(val)
        }, completeFn: {
            completeFn()
        }) { err in
            errorFn(err)
        }
    }
    
    public func mapAny() -> IOMonad<Any> {
        return self.map { val in
            return val as Any
        }
    }
}

extension IOMonad {
    public func filter(_ predicateFn: @escaping Function1<T, Bool>) -> IOMonad {
        return self.bindFunctor {
            return { val in
                if predicateFn(val) {
                    return (val, false)
                }
                return (nil, false)
            }
        }
    }
    
    public func map(_ mapFn: @escaping Function1<T, T>) -> IOMonad {
        return self.bindFunctor {
            return { val in
                return (mapFn(val), false)
            }
        }
    }
    
    public func take(until predicateFn: @escaping Function1<T, Bool>) -> IOMonad {
        return self.bindFunctor {
            return { val in
                if predicateFn(val) {
                    return (nil, true)
                } else {
                    return (val, false)
                }
            }
        }
    }
    
    public func take(_ num: Int) -> IOMonad {
        return self.bindFunctor {
            var count = 0;
            return { val in
                if (count < num) {
                    count = count + 1
                    // 如果当前是最后一个，则告知完成
                    return (val, count == num)
                } else {
                    return (nil, true)
                }
            }
        }
    }
    
    public func skip(_ num: Int) -> IOMonad {
        return self.bindFunctor {
            var count = 0;
            return { val in
                if (count < num) {
                    count = count + 1
                    return (nil, false)
                } else {
                    return (val, false)
                }
            }
        }
    }
    
    public func ignoreNilValue() -> IOMonad {
        return self.bindFunctor {
            return { val in
                let anyValue = val as Any
                if case Optional<Any>.none = anyValue {
                    return (nil, false)
                } else {
                    return (val, false)
                }
            }
        }
    }

    public func ignoreAll() -> IOMonad {
        return self.bindFunctor {
            return { val in
                return (nil, false)
            }
        }
    }
}

extension IOMonad {
    public func flattenMap<U>(_ mapFn: @escaping Function1<T, IOMonad<U>>) -> IOMonad<U> {
        return self.bindMonad {
            return { val in
                return (mapFn(val), false)
            }
        }
    }
    
    /** 对于IOMonad<IOMonad<T>>的展平操作 */
    public func flatten<U>() -> IOMonad<U> where T == IOMonad<U> {
        return self.flattenMap { stream in
            return stream
        }
    }
    
    public func map<U>(_ mapFn: @escaping Function1<T, U>) -> IOMonad<U> {
        return self.flattenMap { val in
            let newVal = mapFn(val)
            return IOMonad<U>.just(newVal)
        }
    }
    
    /** 处理异常，因为catch的目的是为了保证sendNext的执行，所以返回值的类型应该返回一致 */
    public func catchError(_ bindCatchFn: @escaping Function1<Error, IOMonad<T>>) -> IOMonad<T> {
        return IOMonad<T> { thunker in
            let mutex = Mutex()
            let composeDisposer = IODisposer()
            var allDisposer: [IODisposer] = []
            
            // Monad操作默认是值语义
            let preDisposer = self.enableValueSemantics().subscribe(nextFn: { val in
                thunker.sendNext(val)
            }, completeFn: {
                thunker.sendComplete()
            }) { err in
                // 如果未能处理error，则会触发被deliver的thunker.sendError
                let signal = bindCatchFn(err)
                let errDisposer = signal.deliver(to: thunker)
                // signal.deliver可能触发stream终止，为了避免dispose触发的死锁，降低锁的力度
                let succ = mutex.tryLock()
                if composeDisposer.isDisposed {
                    errDisposer.dispose()
                } else {
                    allDisposer.append(errDisposer)
                }
                if succ {
                    mutex.unlock()
                }
            }

            composeDisposer.setDispose {
                let succ = mutex.tryLock()
                for disposer in allDisposer {
                    disposer.dispose()
                }
                preDisposer.dispose()
                if succ {
                    mutex.unlock()
                }
            }

            return composeDisposer.dispose
        }.debugName("catch")
    }
    
    public func concat(_ signal: IOMonad<T>) -> IOMonad<T> {
        return IOMonad { thunker in
            let composeDisposer = IODisposer()
            var currDisposer: IODisposer? = nil
            let preDisposer = self.enableValueSemantics().subscribe(nextFn: { val in
                thunker.sendNext(val)
            }, completeFn: {
                currDisposer = signal.enableValueSemantics().subscribe(nextFn: { val in
                    thunker.sendNext(val)
                }, completeFn: {
                    thunker.sendComplete()
                }) { err in
                    thunker.sendError(err)
                }
            }) { err in
                thunker.sendError(err)
            }
            composeDisposer.setDispose {
                preDisposer.dispose()
                if let currDisposer = currDisposer {
                    currDisposer.dispose()
                }
            }
            return composeDisposer.dispose
        }
    }
    
    public func doNext(nextFn: @escaping Action1<T>) -> IOMonad {
        return self.bindFunctor {
            return { val in
                nextFn(val)
                return (val, false)
            }
        }
    }
    
    public func doComplete(completeFn: @escaping Action) -> IOMonad {
        return IOMonad<T> { thunker in
            return self.enableValueSemantics().subscribe(nextFn: { val in
                thunker.sendNext(val)
            }, completeFn: {
                completeFn()
                thunker.sendComplete()
            }) { err in
                thunker.sendError(err)
            }.dispose
        }
    }
    
    public func doError(errorFn: @escaping Action1<Error>) -> IOMonad {
        return IOMonad<T> { thunker in
            return self.enableValueSemantics().subscribe(nextFn: { val in
                thunker.sendNext(val)
            }, completeFn: {
                thunker.sendComplete()
            }) { err in
                errorFn(err)
                thunker.sendError(err)
            }.dispose
        }
    }
}

extension IOMonad {
    @discardableResult
    public func subscribeNext(_ nextFn: @escaping Action1<T>) -> IODisposer {
        return self.subscribe(nextFn: nextFn)
    }
    
    @discardableResult
    public func subscribeComplete(_ completeFn: @escaping Action) -> IODisposer {
        return self.subscribe(completeFn: completeFn)
    }
    
    @discardableResult
    public func subscribeError(_ errorFn: @escaping Action1<Error>) -> IODisposer {
        return self.subscribe(errorFn: errorFn)
    }
    
    /** 将当前stream的result转发给外部的thunker */
    @discardableResult
    public func deliver(to thunker: IOThunker<T>) -> IODisposer {
       return self.subscribe(nextFn: thunker.sendNext,
                             completeFn: thunker.sendComplete,
                             errorFn: thunker.sendError)
    }
}

///////////////////////////////////////////

public class IOSubject<T> {
    // 为了处理被多次订阅
    private var thunkerCopyList: [IOThunker<T>] = []
    private var stream: IOMonad<T>!
    private var nextFn: Action1<T> = { _ in }
    private var completeFn: Action = { }
    private var errorFn: Action1<Error> = { _ in }
    private let disposeFn: Action
    
    var signal: IOMonad<T> {
        return stream
    }
    
    deinit {
        self.disposeFn()
    }

    init() {
        var isDispose = false
        // 解决同步代码，Subject在销毁后才触发对stream的订阅
        self.disposeFn = {
            isDispose = true
        }
        self.stream = IOMonad<T> {[unowned self] thunker in
            if !isDispose {
                self.thunkerCopyList.append(thunker)
                
                self.nextFn = { val in
                    for item in self.thunkerCopyList {
                        item.sendNext(val)
                    }
                }
                self.completeFn = {
                    for item in self.thunkerCopyList {
                        item.sendComplete()
                    }
                }
                self.errorFn = { err in
                    for item in self.thunkerCopyList {
                        item.sendError(err)
                    }
                }
            }
            return nil
        }.debugName("subject")
        // 实际应用中多次订阅Subject的场景不多
        self.stream.enableValueSemantics()
    }
    
    func sendNext(_ val: T) {
        self.nextFn(val)
    }
    
    func sendComplete() {
        self.completeFn()
    }
    
    func sendError(_ err: Error) {
        self.errorFn(err)
    }
}

extension IOSubject {
    @discardableResult
    public func subscribeNext(_ nextFn: @escaping Action1<T>) -> IODisposer {
        return self.signal.subscribe(nextFn: nextFn)
    }
    
    @discardableResult
    public func subscribeComplete(_ completeFn: @escaping Action) -> IODisposer {
        return self.signal.subscribe(completeFn: completeFn)
    }
    
    @discardableResult
    public func subscribeError(_ errorFn: @escaping Action1<Error>) -> IODisposer {
        return self.signal.subscribe(errorFn: errorFn)
    }
    
    /** 将当前stream的result转发给外部的thunker */
    @discardableResult
    public func deliver(to thunker: IOThunker<T>) -> IODisposer {
       return self.signal.subscribe(nextFn: thunker.sendNext,
                                    completeFn: thunker.sendComplete,
                                    errorFn: thunker.sendError)
    }
}

// REMARK: takeUnitl: rac_willDealloc

extension IOMonad {
    public func take<U: Deinitable>(until stream: IOMonad<U>) -> IOMonad {
        let mutex = Mutex()
        return IOMonad<T> { thunker in
            let composeDisposer = IODisposer()
            let triggerCompleteFn = {
                let succ = mutex.tryLock()
                composeDisposer.dispose()
                thunker.sendComplete()
                if succ {
                    mutex.unlock()
                }
            }
            let triggerDisposer = stream.enableValueSemantics().subscribe(nextFn: { val in
               triggerCompleteFn()
            }, completeFn: {
                triggerCompleteFn()
            })
            
            let succ = mutex.tryLock()
            composeDisposer.setDispose {
                triggerDisposer.dispose()
            }
            if succ {
                mutex.unlock()
            }
            
            if !composeDisposer.isDisposed {
                let preDisposer = self.enableValueSemantics().subscribe(nextFn: { val in
                    thunker.sendNext(val)
                }, completeFn: {
                    thunker.sendComplete()
                    let succ = mutex.tryLock()
                    composeDisposer.dispose()
                    if succ {
                        mutex.unlock()
                    }
                }) { err in
                    thunker.sendError(err)
                }
                
                let succ = mutex.tryLock()
                composeDisposer.setDispose {
                    triggerDisposer.dispose()
                    preDisposer.dispose()
                }
                if succ {
                    mutex.unlock()
                }
            }
            
            return composeDisposer.dispose
        }.enableValueSemantics()
    }
}
